package org.util.mqttclient;

import java.net.URISyntaxException;

import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.CallbackConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;

public class TopicListener {
	private MQTT mqtt;

	public TopicListener(String host, int port, String username, String password)
			throws URISyntaxException {
		mqtt = new MQTT();
		mqtt.setHost(host, port);
		mqtt.setUserName(username);
		mqtt.setPassword(password);
		mqtt.setKeepAlive((short)1);
	}
	
	public void onPublish(final String[] topicNames,final ReceivedCallback receivedCallback) throws InterruptedException{
		 final CallbackConnection connection = mqtt.callbackConnection();
	        
	        connection.listener(new org.fusesource.mqtt.client.Listener() {

	            public void onConnected() {
	            }
	            public void onDisconnected() {
	            }
	            public void onFailure(Throwable value) {
	                value.printStackTrace();
	            }
	            public void onPublish(UTF8Buffer topic, Buffer msg, Runnable ack) {
	                String body = msg.utf8().toString();
	                receivedCallback.receivedHandler(body);
	            }
	        });
	        connection.connect(new Callback<Void>() {
	            @Override
	            public void onSuccess(Void value) {
	                Topic[] topics = new Topic[topicNames.length];
	                for(int i=0;i<topicNames.length;i++){
	                	topics[i]=new Topic(topicNames[i], QoS.EXACTLY_ONCE); 
	                }
	                connection.subscribe(topics, new Callback<byte[]>() {
	                    public void onSuccess(byte[] qoses) {
	                    }
	                    public void onFailure(Throwable value) {
	                        value.printStackTrace();
	                    }
	                });
	            }
	            @Override
	            public void onFailure(Throwable value) {
	                value.printStackTrace();
	                System.exit(1);
	            }
	        });
	        // Wait forever..
	}
}
